package org.easy.async;

import cn.hutool.core.thread.NamedThreadFactory;

import java.util.Map;
import java.util.Queue;
import java.util.concurrent.*;

/**
 * @Description:
 * @Author: wxl
 * @CreateTime: 2023/1/31 17:15
 */
public abstract class EasyQueue<T> {

    private final static int CORE_POOL_SIZE = 2;

    private final static int MAX_POOL_SIZE  = 10;

    private final static int KEEP_ALIVE_TIME = 0;

    private final static int WORK_QUEUE_SIZE = 50;

    protected final Map<String, T> objCacheMap = new ConcurrentHashMap<>();

    protected final Queue<T> queue = new LinkedBlockingDeque<>();

    protected final RejectedExecutionHandler handler = initRejectedExecutionHandler();

    protected final ThreadPoolExecutor queueThreadPool = new ThreadPoolExecutor(CORE_POOL_SIZE,MAX_POOL_SIZE,KEEP_ALIVE_TIME, TimeUnit.SECONDS,new ArrayBlockingQueue<>(WORK_QUEUE_SIZE),this.handler);

    public EasyQueue () {
        NamedThreadFactory namedThreadFactory = new NamedThreadFactory("execute-pool-file-",true);
        this.queueThreadPool.setThreadFactory(namedThreadFactory);
    }

    final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(5);

    final ScheduledFuture<?> scheduledFuture = scheduler.scheduleAtFixedRate(() -> {
        // 判断缓冲队列是否存在记录
        if (!queue.isEmpty()) {
            // 当前线程的队列容量
            if (queueThreadPool.getQueue().size() < WORK_QUEUE_SIZE) {
                T object = queue.poll();
                executeObject(object);
            }
        }
    },0,1,TimeUnit.SECONDS);

    /**
     * 添加
     * @param object 对象
     */
    public void addObject(T object,String key) {
        objCacheMap.putIfAbsent(key,object);
        executeObject(object);
    }

    protected void executeObject (T object) {
        EasyThread<T> tEasyThread = new EasyThread<>(object);
        queueThreadPool.execute(tEasyThread);
    }


    /**
     * 根据不同的实现不同的拒绝策略
     * @return RejectedExecutionHandler
     */
    public abstract RejectedExecutionHandler initRejectedExecutionHandler();

    public Queue<T> getQueue () {
        return this.queue;
    }

    public void shutdown () {
        scheduler.shutdown();
        queueThreadPool.shutdown();
    }

}
